package spring.boot.rocketmq.receiver.consumer;

import lombok.extern.log4j.Log4j2;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationListener;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.event.ContextRefreshedEvent;
import spring.boot.rocketmq.receiver.config.TopicOrderConsumerConfigure;
import spring.boot.rocketmq.receiver.config.TopicTestConsumerConfigure;

import java.io.UnsupportedEncodingException;
import java.util.List;

@Log4j2
@Configuration
public class OrderTopicConsumer extends TopicOrderConsumerConfigure implements ApplicationListener<ContextRefreshedEvent> {
    protected final Logger logger = LoggerFactory.getLogger(this.getClass());
    @Override
    public void onApplicationEvent(ContextRefreshedEvent arg0) {
        try {
            super.listener("TopicOrder", "*");
        } catch (MQClientException e) {
            log.error("TopicOrder 消费者监听器启动失败", e);
        }

    }

    @Override
    public ConsumeConcurrentlyStatus dealBody(List<MessageExt> msgs)  {
        for(MessageExt msg : msgs) {
            try {
                String msgStr = new String(msg.getBody(), "utf-8");
                logger.info("TopicOrder 第{}次消费，ID:{} KEYS:{}  message:{}",msg.getReconsumeTimes(),msg.getMsgId(),msg.getKeys(),msgStr);
            } catch (UnsupportedEncodingException e) {
                logger.error("TopicOrder body转字符串解析失败");
            }
        }
        //消费失败  默认重试16次  10s;30s;1m;2m;3m......
        //return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        //消费完成
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
}
